package com.atguigu.bigdata.edu.realtime.app.dwd;

import com.atguigu.bigdata.edu.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.edu.realtime.common.Constant;
import com.atguigu.bigdata.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class Dwd_05_DwdTradePayDetailSuc extends BaseSQLApp {

    public static void main(String[] args) {

        new Dwd_05_DwdTradePayDetailSuc().init(3555, 2, "Dwd_05_DwdTradePayDetailSuc");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //读取ods_db
        readOdsDb(tEnv, "Dwd_05_DwdTradePayDetailSuc");

        //从ods_db过滤出 支付成功表数据
        Table paymentInfo = tEnv.sqlQuery("SELECT " +
                "\tdata['id'] id, " +
                "\tdata['order_id'] order_id, " +
                "\tdata['payment_type'] payment_type, " +
                "\tdata['callback_time'] callback_time, " +
                "\tts " +
                "FROM ods_db " +
                "WHERE `database` = 'edu' " +
                "and  `table` = 'payment_info' " +
                "and `type` = 'insert' " +
                "and data['payment_status'] = '1602' ");

        //paymentInfo.execute().print();
        tEnv.createTemporaryView("payment_info", paymentInfo);
        //读取下单详情表
        tEnv.executeSql("create table dwd_trade_order_detail(  " +
                "id string,  " +
                "order_id string,  " +
                "user_id string,  " +
                "course_id string,  " +
                "course_name string,  " +
                "province_id string,  " +
                "date_id string,  " +
                "create_time string,  " +
                "origin_amount string,  " +
                "final_amount string,  " +
                "session_id string,  " +
                "out_trade_no string,  " +
                "ts bigint,  " +
                "row_op_ts timestamp_ltz(3)" +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_05_DwdTradePayDetailSuc"));

        //2张表进行Join
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        Table result = tEnv.sqlQuery("select " +
                "od.id order_detail_id, " +
                "od.order_id, " +
                "od.user_id, " +
                "od.course_id, " +
                "od.course_name, " +
                "od.province_id, " +
                "pi.payment_type payment_type_code, " +
                "pi.callback_time, " +
                "od.origin_amount, " +
                "od.final_amount, " +
                "od.session_id, " +
                "od.out_trade_no, " +
                "od.date_id, " +
                "pi.ts, " +
                "od.row_op_ts row_op_ts " +
                "from dwd_trade_order_detail od " +
                "join payment_info pi on od.order_id = pi.order_id" );
         //result.execute().print();
        //写到kafka中
        tEnv.executeSql("create table dwd_trade_pay_detail_suc( " +
                "order_detail_id string, " +
                "order_id string, " +
                "user_id string, " +
                "course_id string, " +
                "course_name string, " +
                "province_id string, " +
                "payment_type_code string, " +
                "callback_time string, " +
                "origin_amount string, " +
                "final_amount string, " +
                "session_id string, " +
                "out_trade_no string, " +
                "date_id string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));

       result.executeInsert("dwd_trade_pay_detail_suc");
    }
}